package spring.cloud.tasks.executor.name_space;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import spring.cloud.tasks.common.CuratorFrameworkUtils;
import spring.cloud.tasks.executor.utils.LogUtils;
import spring.cloud.tasks.executor.utils.SystemEnvProperties;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

/**
 * 基于Zookeeper的注册中心.
 */
@Slf4j
@Setter
@Getter
public class NameSpaceDataManagerImpl implements NameSpaceDataManager {

    private static final String SLASH = "/";
    /**
     * 最小连接超时时间
     */
    private static final int MIN_CONNECTION_TIMEOUT = 20 * 1000;
    /**
     * 最小会话超时时间
     */
    private static final int MIN_SESSION_TIMEOUT = 20 * 1000;
    private NameSpaceConfiguration nameSpaceConfiguration;
    /**
     * 会话超时时间
     */
    private int sessionTimeoutMs;
    private CuratorFramework curatorFramework;
    private String executorId;

    public NameSpaceDataManagerImpl(final NameSpaceConfiguration nameSpaceConfiguration) {
        this.nameSpaceConfiguration = nameSpaceConfiguration;
    }

    private static int calculateConnectionTimeoutMs() {
        // default SystemEnvProperties._ZK_CLIENT_CONNECTION_TIMEOUT_IN_SECONDS = -1
        int connectionTimeoutMs = SystemEnvProperties.ZK_CLIENT_CONNECTION_TIMEOUT_IN_SECONDS * 1000;
        return connectionTimeoutMs > MIN_CONNECTION_TIMEOUT ? connectionTimeoutMs : MIN_CONNECTION_TIMEOUT;
    }

    private static int calculateSessionTimeoutMs() {
        // default SystemEnvProperties._ZK_CLIENT_SESSION_TIMEOUT_IN_SECONDS = -1
        int sessionTimeoutMs = SystemEnvProperties.ZK_CLIENT_SESSION_TIMEOUT_IN_SECONDS * 1000;
        return sessionTimeoutMs > MIN_SESSION_TIMEOUT ? sessionTimeoutMs : MIN_SESSION_TIMEOUT;
    }

    @Override
    public String getNameSpace() {
        return nameSpaceConfiguration.getNameSpace();
    }

    @Override
    public void init() {
        curatorFramework = buildCuratorFramework();
        curatorFramework.start();

        try {
            curatorFramework.getZookeeperClient().blockUntilConnectedOrTimedOut();
            if (!curatorFramework.getZookeeperClient().isConnected()) {
                throw new RuntimeException("the zk client is not connected while reach connection timeout");
            }
            // check nameSpace node by using client, for UnknownHostException of connection string.
            curatorFramework.checkExists().forPath(SLASH + nameSpaceConfiguration.getNameSpace());
            // CHECKSTYLE:OFF
        } catch (final Exception ex) {
            throw new RuntimeException("zk connect fail, zkList is " + nameSpaceConfiguration.getConnectString(), ex);
        }

        LogUtils.info(log, "", "zkClient is created successfully.");
    }

    private CuratorFramework buildCuratorFramework() {
        /**
         * 重试机制单位毫秒
         */
        ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(//
                nameSpaceConfiguration.getBaseSleepTimeMs(), //
                nameSpaceConfiguration.getMaxRetries(), //
                nameSpaceConfiguration.getMaxSleepMs());//
        Builder builder = CuratorFrameworkFactory.builder()//
                .connectString(nameSpaceConfiguration.getConnectString())//
                .namespace(nameSpaceConfiguration.getNameSpace())//
                .retryPolicy(exponentialBackoffRetry);
        int connectionTimeoutMs;
        if (0 != nameSpaceConfiguration.getConnectionTimeoutMs()) {
            connectionTimeoutMs = nameSpaceConfiguration.getConnectionTimeoutMs();
        } else {
            connectionTimeoutMs = calculateConnectionTimeoutMs();
        }
        builder.connectionTimeoutMs(connectionTimeoutMs);
        if (0 != nameSpaceConfiguration.getSessionTimeoutMs()) {
            sessionTimeoutMs = nameSpaceConfiguration.getSessionTimeoutMs();
        } else {
            sessionTimeoutMs = calculateSessionTimeoutMs();
        }
        builder.sessionTimeoutMs(sessionTimeoutMs);
        if (!Strings.isNullOrEmpty(nameSpaceConfiguration.getDigest())) {
            builder.authorization("digest", nameSpaceConfiguration.getDigest().getBytes(StandardCharsets.UTF_8))
                    .aclProvider(new ACLProvider() {
                        @Override
                        public List<ACL> getDefaultAcl() {
                            return ZooDefs.Ids.CREATOR_ALL_ACL;
                        }

                        @Override
                        public List<ACL> getAclForPath(final String path) {
                            return ZooDefs.Ids.CREATOR_ALL_ACL;
                        }
                    });
        }
        LogUtils.info(log, "", "zookeeper registry center init, server lists is: {}, connection_timeout: {}, session_timeout: {}, retry_times: {}", nameSpaceConfiguration.getConnectString(), connectionTimeoutMs, sessionTimeoutMs, nameSpaceConfiguration.getMaxRetries());
        return builder.build();
    }

    @Override
    public boolean isConnected() {
        return curatorFramework != null && curatorFramework.getZookeeperClient().isConnected();
    }

    @Override
    public CuratorFramework getCuratorFramework() {
        return curatorFramework;
    }

    @Override
    public void close() {
        CloseableUtils.closeQuietly(curatorFramework);
    }

    @Override
    public String getExecutorId() {
        return executorId;
    }

    //
    @Override
    public void setExecutorId(String executorId) {
        this.executorId = executorId;
    }

    //
    @Override
    public void addConnectionStateListener(final ConnectionStateListener connectionStateListener) {
        curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
    }

    @Override
    public void removeConnectionStateListener(final ConnectionStateListener connectionStateListener) {
        curatorFramework.getConnectionStateListenable().removeListener(connectionStateListener);
    }

    //
    @Override
    public void createIfNeedOrSetData(final String path, final String value) {
        try {
            if (!checkExists(path)) {
                curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, value.getBytes());
            } else {
                setData(path, value);
            }
        } catch (final Exception ex) {
        }
    }

    @Override
    public void setData(final String path, final String value) {
        try {
            curatorFramework.inTransaction().check().forPath(path).and().setData().forPath(path, value.getBytes(StandardCharsets.UTF_8)).and().commit();
        } catch (final Exception ex) {
        }
    }

    @Override
    public void createEphemeralDataNode(final String path, final String value) {
        try {
            if (checkExists(path)) {
                curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
            }
            curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, value.getBytes(StandardCharsets.UTF_8));
        } catch (final Exception ex) {
        }
    }

    @Override
    public void createEphemeralSequentialDataNode(final String path) {
        try {
            curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
        } catch (final Exception ex) {
        }
    }

    @Override
    public void delete(final String path) {
        try {
            curatorFramework.delete().guaranteed().forPath(path);
        } catch (KeeperException.NotEmptyException e) {
            LogUtils.debug(log, "", "try to delete path: {} but fail for NotEmptyException", path, e);
            try {
                CuratorFrameworkUtils.deleteRecursive(curatorFramework, path);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        } catch (KeeperException.NoNodeException e) {
            LogUtils.debug(log, "", "try to delete path: {} but fail for NoNodeException", path, e);
        } catch (final Exception ex) {

        }
    }


    @Override
    public boolean checkExists(final String path) {
        try {
            return null != curatorFramework.checkExists().forPath(path);
        } catch (final Exception ex) {
            return false;
        }
    }

    @Override
    public String getData(final String path) {
        try {
            byte[] bytes = curatorFramework.getData().forPath(path);
            if (bytes == null) {
                return null;
            }
            return new String(bytes, StandardCharsets.UTF_8);
        } catch (final Exception ex) {
            return null;
        }
    }

    @Override
    public long getCtime(final String path) {
        long ctime = 0L;
        try {
            String $path = curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
            ctime = curatorFramework.checkExists().forPath($path).getCtime();
        } catch (final Exception ex) {

        }
        Preconditions.checkState(0L != ctime, "Cannot get registry center time.");
        return ctime;
    }

    @Override
    public Stat getStat(final String path) {
        try {
            return curatorFramework.checkExists().forPath(path);
        } catch (final Exception ex) {
            return null;
        }
    }

    @Override
    public List<String> getChildrenList(final String path) {
        List<String> list = null;
        try {
            list = curatorFramework.getChildren().forPath(path);
            Collections.sort(list, new Comparator<String>() {
                @Override
                public int compare(final String o1, final String o2) {
                    return o2.compareTo(o1);
                }
            });
            return list;
        } catch (final Exception ex) {
            return Collections.emptyList();
        }
    }


}
